package com.cl.spark.node;

import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MatchNode extends BaseSparkNode {
    @Autowired
    SparkSession sparkSession;

    @Override
    public SparkResult process(SparkParam sparkParam) {
        JSONObject expression = sparkParam.getNodeExpression();

        Dataset<Row> dataset1 = sparkParam.getSparkResultList().get(0).getDataset();
        Dataset<Row> dataset2 = sparkParam.getSparkResultList().get(1).getDataset();

        String field1 = expression.getJSONArray("fields1").getString(0);
        String field2 = expression.getJSONArray("fields2").getString(0);

        String outputType = expression.getString("outputType");
        String matchType = expression.getString("matchType");
        String condition = "d1." + field1 + " = d2." + field2;
        if ("模糊匹配".equals(matchType)) {
            condition = "d1." + field1 + " LIKE CONCAT('%', d2." + field2 + ", '%')";
        }
        String sql = "SELECT d1.*, CASE WHEN " + condition + " THEN true ELSE false END AS MATCH FROM dataset1 d1 LEFT JOIN dataset2 d2 ON d1.id = d2.id";
        if ("匹配".equals(outputType)) {
            sql = "SELECT d1.*,true AS MATCH FROM dataset1 d1 INNER JOIN dataset2 d2 ON d1.id = d2.id AND "+ condition;
        } else if ("不匹配".equals(outputType)) {
            sql = "SELECT d1.*,false AS MATCH FROM dataset1 d1 LEFT ANTI JOIN dataset2 d2 ON d1.id = d2.id  AND " + condition;

        }
        // 注册临时视图以便使用 SQL 查询
        dataset1.createOrReplaceTempView("dataset1");
        dataset2.createOrReplaceTempView("dataset2");

        // 执行 SQL 查询来比较字段值并添加 MATCH 字段
        Dataset<Row> result = sparkSession.sql(sql);
        return SparkResult.success(result);
    }


    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.MATCH;
    }
}
